package com.atguigu.dw.gmall.realtime.app

import java.text.SimpleDateFormat
import java.util
import java.util.Date

import com.alibaba.fastjson.JSON
import com.atguigu.constan.ConstanVal
import com.atguigu.dw.gmall.realtime.bean.{AlertInfo, EventLog}
import com.atguigu.dw.gmall.realtime.util.{EsUtil, MyKafkaUtil}
import org.apache.spark.streaming.{Seconds, StreamingContext}


/**
 * description ：预警 app
 * author      ：剧情再美终是戏 
 * mail        : 13286520398@163.com
 * date        ：Created in 2020/2/14 15:30
 * modified By ：
 * version:    : 1.0
 */
object AlterApp extends App {

  private val format = new SimpleDateFormat("yyyy-MM-dd HH:mm")

  override def handle(ssc: StreamingContext) = {
    // 读取 kafkaStream 数据
    val dawStream = MyKafkaUtil.getKafkaStream(ssc, Set[String](ConstanVal.KAFKA_TOPIC_EVENT)).window(Seconds(60))

    // 将 kafkaStream 数据转换为样例类
    val dawBean = dawStream.map({
      case (_, v) =>
        val log = JSON.parseObject(v, classOf[EventLog])
        (log.mid, log)
    })

    /*
    1.同一设备(按照设备分组)
    2.5分钟内(window)
    3.三个不同账号登录
    4.领取优惠券
    5.并且没有浏览商品
    6.同一设备每分钟只预警一次(同一设备, 每分钟只向 es 写一次记录)
     */
    val alterInfoStream = dawBean
      .groupByKey
      .map({
        case (mid, logIt) => {
          // 记录登录账号
          val uidSet = new util.HashSet[String]()
          // 记录所有的操作事务
          val eventList = new util.ArrayList[String]()
          // 记录优惠券应用的商品id
          val itemSet = new util.HashSet[String]()
          // 判断是否有点击行为
          var isClickItem: Boolean = false

          // 循环处理数据
          import scala.util.control.Breaks._
          breakable {
            for (elem <- logIt) {
              val eventId = elem.eventId // 点击行为
              eventList.add(eventId)
              eventId match {
                case "coupon" =>
                  uidSet.add(elem.uid)
                  itemSet.add(elem.itemId)
                case "clickItem" =>
                  isClickItem = true
                  break
                case _ => // nothing to to
              }
            }
          }
          (!isClickItem && uidSet.size() >= 3, AlertInfo(mid, uidSet, itemSet, eventList, System.currentTimeMillis()))
        }
      })


    // 筛选需要预警的数据，并插入 es 中
    alterInfoStream
      .filter(_._1)
      .map {
        case (_, alterInfo) => (alterInfo, alterInfo.mid + "_" + format.format(new Date(alterInfo.ts)))
      }
      .foreachRDD {
        rdd =>
          rdd.foreachPartition(rddIt => EsUtil.insertBulk(ConstanVal.ES_INDEX_GMALL0830_COUPON_ALERT, rddIt))
      }

    alterInfoStream.print(1000)
  }
}

